package com.chenxu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.chenxu.gmall.realtime.common.GmallConfig;
import com.chenxu.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * Date: 2021/07/14
 * Desc:  写出维度数据的Sink实现类
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    // 定义Phoenix连接对象，提高作用域；这个连接为什么不能直接放入BaseDBAPP中，因为会报一个序列化错误；
    // 类似Spark中，主程序是在DRIVER中执行，算子执行在manager中执行，有一个传输过程，但conn没有序列化，所以会报错；
    private Connection conn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        //对连接对象进行初始化
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /**
     * 对流中的数据进行处理
     * @param jsonObj
     * @param context
     * @throws Exception
     */
    //对每条数据调用；
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取目标表的名称
        String tableName = jsonObj.getString("sink_table");
        //获取json中data数据   data数据就是经过过滤之后  保留的业务表中字段
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");

        if (dataJsonObj != null && dataJsonObj.size() > 0) {
            //根据data中属性名和属性值  生成upsert语句（phoenix的语法）
            //phoenix会把表名转化为大写，这里也同样转化一下；
            String upsertSql = genUpsertSql(tableName.toUpperCase(), dataJsonObj);
            System.out.println("向Phoenix插入数据的SQL:" + upsertSql);

            //执行SQL
            PreparedStatement ps = null;
            try {
                ps = conn.prepareStatement(upsertSql);
                //ps.executeQuery()主要是用来查询语句，返回查询结果；
                //ps.executeUpdate()主要是用来做dml语句的，主要是增删改，返回的是int类型，代表影响的记录条数；
                ps.execute();
                //注意：执行完Phoenix插入操作之后，需要手动提交事务
                //自动事务提交被关闭了，需要手动提交；
                conn.commit();
            } catch (SQLException e) {
                e.printStackTrace();
                throw new RuntimeException("向Phoenix插入数据失败");
            } finally {
                if (ps != null) {
                    //同样，这里只关闭ps，不关闭conn；
                    ps.close();
                }
            }

            //如果当前做的是更新操作，需要将Redis中缓存的数据清除掉
            if(jsonObj.getString("type").equals("update")){
                DimUtil.deleteCached(tableName,dataJsonObj.getString("id"));
            }
        }


    }

    // 根据data属性和值  生成向Phoenix中插入数据的sql语句
    private String genUpsertSql(String tableName, JSONObject dataJsonObj) {
        /*
        dataJsonObj的类型大致如下所示：
            {
                "id":88,
                "tm_name":"xiaomi"
            }
        */
        //"upsert into 表空间.表名(列名.....) values (值....)"
        //按照例子，key就是id和tm_name;value就是88和xiaomi；
        Set<String> keys = dataJsonObj.keySet();
        Collection<Object> values = dataJsonObj.values();
        String upsertSql = "upsert into " + GmallConfig.HABSE_SCHEMA + "." + tableName + "(" +
            StringUtils.join(keys, ",") + ")";

        //注意写法、空格等等；由于之前建表的时候，都是varchar类型，所以这里都要用'括起来'；
        String valueSql = " values ('" + StringUtils.join(values, "','") + "')";
        //效果类似：values ('10','zs');
        return upsertSql + valueSql;
    }
}
